package cn.xiao.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCount {

    // HDFS的路径
    private static String path = "/words";

    // Linux路径
    private static String outPath = "/wcountTest";



    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        // 构建Job对象
        Job job = Job.getInstance(new Configuration());

        // 注意：一定要将Main方法写进去
        job.setJarByClass(WordCount.class);

        // 这里是设置mapper的输出类型【对应WCMapper的context.write】
        job.setMapperClass(WCMapper.class);
        // 输出key类型
        job.setOutputKeyClass(Text.class);
        // 输出value类型
        job.setMapOutputValueClass(LongWritable.class);

        // 这里是hadoop的path
        FileInputFormat.setInputPaths(job, new Path(path));


        // 这里是设置reduce的输出类型【对应Reduce的context.write】
        job.setReducerClass(WCReducer.class);
        // 输出key类型
        job.setOutputKeyClass(Text.class);
        // 输出value类型
        job.setOutputValueClass(LongWritable.class);

        // 这里的包是org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
        FileOutputFormat.setOutputPath(job, new Path(outPath));

        // 提交任务，执行
        job.waitForCompletion(true); // true 执行过程中，打印进度


//        Job job = Job.getInstance(new Configuration());
//
//        //notice
//        job.setJarByClass(WordCount.class);
//
//        //set mapper`s property
//        job.setMapperClass(WCMapper.class);
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(LongWritable.class);
//        FileInputFormat.setInputPaths(job, new Path(path));
//
//        //set reducer`s property
//        job.setReducerClass(WCReducer.class);
//        job.setOutputKeyClass(Text.class);
//        job.setOutputValueClass(LongWritable.class);
//        FileOutputFormat.setOutputPath(job, new Path(outPath));
//
//        //submit
//        job.waitForCompletion(true);

    }





}
